package fun.tan90.easy.binlog.manager.event;

import cn.hutool.core.lang.Tuple;
import com.mybatisflex.core.row.Db;
import fun.tan90.easy.binlog.base.constants.EasyBinlogConstants;
import fun.tan90.easy.binlog.base.threadpool.EasyBinlogThreadPool;
import fun.tan90.easy.binlog.manager.service.BinlogWorkerService;
import lombok.extern.slf4j.Slf4j;
import org.noear.dami.Dami;
import org.noear.socketd.transport.core.Session;
import org.springframework.stereotype.Component;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.io.IOException;
import java.util.Date;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;
import java.util.stream.Stream;

/**
 * description TODO
 * company 铁人科技
 *
 * @author 陈浩杰
 * @date 2023/12/18 16:43
 */
@Slf4j
@Component
public class DamiBusEvent {

    public final static String RECORD_WORKER_EVENT_RECORD = "record.worker.event.record";

    public final static String WORKER_OFFLINE = "worker.offline";

    public final static String WORKER_ONLINE = "worker.online";

    public final static String OUTPUT_CHANGE = "output.change";

    @Resource
    BinlogWorkerService binlogWorkerService;

    @PostConstruct
    public void damiBusEvent() {
        Dami.<String, Void>bus().listen(WORKER_OFFLINE, payload -> {
            String workId = payload.getContent();
            Db.updateBySql("update binlog_master set slave_work_id = null, slave_open_res = null, slave_client_status = '0' where slave_work_id = ?", workId);
        });

        Dami.<Session, Void>bus().listen(WORKER_ONLINE, payload -> {
            EasyBinlogThreadPool.EB_SCHEDULED_EXECUTOR.schedule(() -> {
                Session session = payload.getContent();
                String workId = session.param("workId");
                try {
                    String masterCodes = session.sendAndRequest(EasyBinlogConstants.WORKER_RUNNING_MASTER_CODE, null).await().dataAsString();
                    if (StringUtils.hasLength(masterCodes)) {
                        for (String masterCode : Stream.of(masterCodes.split(",")).collect(Collectors.toSet())) {
                            Db.updateBySql("update binlog_master set slave_work_id = ?, slave_open_res = 'ok', slave_client_status = '1' where master_code = ?", workId, masterCode);
                        }
                    }
                } catch (IOException e) {
                    log.error("get.binlog.master_code发生异常，{}", e.getMessage());
                }
            }, 5, TimeUnit.SECONDS);
        });

        Dami.<Tuple, Void>bus().listen(RECORD_WORKER_EVENT_RECORD, payload -> {
            Tuple tuple = payload.getContent();
            Db.updateBySql("insert into binlog_worker_event_record (work_id, event_type, event_content, create_time) value (?, ?, ?, ?)", tuple.get(0), tuple.get(1), tuple.get(2), new Date());
        });

        Dami.<String, Void>bus().listen(OUTPUT_CHANGE, payload -> {
            EasyBinlogThreadPool.EB_FIXED_EXECUTOR.execute(() -> {
                String outputCode = payload.getContent();
                binlogWorkerService.broadcastSend(EasyBinlogConstants.DELETE_OUTPUT_PARAM, outputCode, null);
            });
        });
    }
}
